package com.dtkavin.akka

import java.util.UUID

import akka.actor.{Actor, ActorSystem, Props}
import com.typesafe.config.ConfigFactory

/**
  * Created by Administrator on 2016/4/9 0009.
  */
class Master(val host: String, val port: Int) extends Actor {
  val LOST_WORKET = 10000
  val masterId: String = UUID.randomUUID().toString

  val workerMap = scala.collection.mutable.HashMap[String, WorkerInfo]()

  override def preStart(): Unit = {
    println("master starting ...")
    import scala.concurrent.duration._
    import scala.concurrent.ExecutionContext.Implicits.global
    context.system.scheduler.schedule(0 millis ,5000 millis,self ,WorkerCheck )
  }


  override def receive: Receive = {
    case HeartBeat(workerId: String, workerBeatTime: Long) => {
      workerMap.get(workerId).get.lastHeartBeatTime = workerBeatTime
    }
    case WorkerCheck => {
     workerMap.filter(System.currentTimeMillis - _._2.lastHeartBeatTime > LOST_WORKET).foreach(workerMap -= _._1)
      println(workerMap.size)
    }
    case WorkderRegister(workerId, host, port, memory, core) => {
      if (!workerMap.contains(workerId)) {
        val worker = new WorkerInfo(workerId, host, port, memory, core,System.currentTimeMillis())
        workerMap(workerId) = worker
        println(workerId)
        sender ! RegisterBack(masterId)
      }
    }
    case _ => println("master received mismathed msg")
  }
}

object Master {
  def apply(host: String, port: Int) = new Master(host, port)

  def main(args: Array[String]) {
    val host = "127.0.0.1"
    val port = 9900

    val confStr = s""" akka.actor.provider = "akka.remote.RemoteActorRefProvider"
                      |akka.remote.netty.tcp.hostname = "$host"
                      |akka.remote.netty.tcp.port = "$port"""" stripMargin

    val conf = ConfigFactory.parseString(confStr)
    val masterSystem = ActorSystem.apply("MasterSystem", conf)
    val master = masterSystem.actorOf(Props(Master(host, port)), "Master")
//        master ! "test"
    masterSystem.awaitTermination()
  }
}
